package cn.tang.tframe.mq.kafka.producer;

import cn.tang.tframe.common.base.MainConfigLoader;
import cn.tang.tframe.common.utils.io.PropertiesUtil;
import cn.tang.tframe.common.utils.lang.SpiUtils;
import cn.tang.tframe.common.utils.lang.StringUtils;
import cn.tang.tframe.dynconfig.DynConfigRegister;
import cn.tang.tframe.dynconfig.IDynConfigProcessor;
import cn.tang.tframe.mq.IQtMessager;
import cn.tang.tframe.mq.producer.IQtProducer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Properties;

/**
 * Created by tanghc on 17/1/20.
 */
public class KafkaProducerClient {
	static Logger logger = LoggerFactory.getLogger(KafkaProducerClient.class);
	private KafkaProducerClient(){}
	private static class KafkaProducerClientHolder{
		private static DelegateKafkaProducer instance;
		private static final String producer_config_group = "common";
		private static final String producer_config_dataId = "kafkaProducer";
		private static final String productCode =
				MainConfigLoader.getInstance().getProductCode();
		private static final String appCode =
				MainConfigLoader.getInstance().getAppCode();
		static {
			DynConfigRegister.getInstance().buildOnce(productCode, appCode, producer_config_group, producer_config_dataId, new IDynConfigProcessor() {
				@Override
				public void procesConfig(String config) {
					Properties properties = PropertiesUtil.loadFromString(config);
					DelegateKafkaProducer delegateKafkaProducer = buildDelegateKafkaProducer(properties);
					if(null != delegateKafkaProducer){
						instance = delegateKafkaProducer;
					}else{
						logger.warn(
								"didnot modify kafka build client is null:productCode={},appCode={}",
								productCode, appCode);
					}
				}
			});
		}

		/**
		 * 属性基于0.10.1
		 */
		private static DelegateKafkaProducer buildDelegateKafkaProducer(Properties properties){
			if(null == properties){
				properties = new Properties();
			}
			//host/port对的列表，用来建立与kafka的初始链接。host1:port1,host2,port2,....
//			setDefault(properties,"bootstrap.servers","");
//			更新metadata的时间间隔，无论partition的leader是否发生变换或者topic其它的元数据是否发生变化。
//			setDefault(properties,"metadata.max.age.ms","");//defalult: 300000
//			producer尝试批量处理消息，可以使用较少的发送次数发送相同数量的消息。
// 可以提高server以及client端的性能。此值是指默认批量处理的字节数。
//			setDefault(properties,"batch.size","");//defalult: 16384
//			producer要求在leader在判定某条消息是否写入完成之前需要收到的确认写入的个数
//			setDefault(properties,"acks","all");//defalult:1
//			该设置通过添加人为的延迟实现这一点：即，请求来到不是立即发送，而是等待指定的延迟，批量发送请求。
//			setDefault(properties,"linger.ms","");//defalult:0
//			请求server时传送给server的clientid字符串。
			setDefault(properties,"client.id",productCode+"_"+appCode);//defalult:""
//			TCP发送的socket的SO_SNDBUF缓存。如果设置为－1，将使用OS的默认值
//			setDefault(properties,"send.buffer.bytes","");//defalult:131072
//			TCP接受缓存的大小（SO_RCVBUF）。如果设置为－1，则使用OS默认值
//			setDefault(properties,"receive.buffer.bytes","");//defalult:32768
//			请求的最大字节数
//			setDefault(properties,"max.request.size","");//defalult:1048576
//			重连给定host之前的等待时间。避免频繁的重连某个host。
//          这个backoff时间也设定了consumer请求broker的重试等待时间。
//			setDefault(properties,"reconnect.backoff.ms","");//defalult:50
//			KafkaProducer.send（）和KafkaProducer.partitionsFor（）将会阻塞的时长
//			setDefault(properties,"max.block.ms","");//defalult:33554432
//			producer用于缓存发送数据的内存大小。如果消息放入缓存的速度大于发送的速度，
//          则producer可以设置阻塞超时时间max.block.ms，超时则报异常即可
//			setDefault(properties,"buffer.memory","");//defalult:33554432
//			重新发送失败请求的等待时间。避免某些失败情况下频繁发送请求。
//			setDefault(properties,"retry.backoff.ms","");//defalult:100
//			producer可以支持的数据压缩类型。合法的压缩格式为：none，gzip，snappy，lz4.
//			setDefault(properties, "compression.type", "");//defalult:none
//			度量样本的计算的时长
//			setDefault(properties,"metrics.sample.window.ms","");//defalult:30000
//          维护计算指标的样本数
//			setDefault(properties,"metrics.num.samples","");//defalult:2
//			用于实现指标统计的类的列表。
//			setDefault(properties,"metric.reporters","");//defalult:[]
//			在阻塞之前，客户端可以在单个链接之中发送未确认的请求的最大数目。
//          注意：如果这个值大于1，则一旦发送失败，有可能会打乱消息的原有顺序
//			setDefault(properties, "max.in.flight.requests.per.connection", "");//defalult:5
//			设置重试次数可以在发送失败时进行重试，提高发送的可靠性。
//			setDefault(properties, "retries", "");//defalult:0
//			空闲链接存在的最长时间，超出之后就会被关闭
//			setDefault(properties,"connections.max.idle.ms","100");//defalult:540000
//			实现Partitioner的接口的Partitioner类
//			setDefault(properties,"partitioner.class","");//defalult:
//			请求的超时时间，即等待server端应答某个请求的最长时间
//			setDefault(properties, "request.timeout.ms", "");//defalult:30000
			/**
			 * 以下参数不允许配置
			 */
//			Serializer接口的密钥的类的key
			overrideValue(properties, "key.serializer", "cn.tang.tframe.mq.kafka.serial.QtMessageSerializer");//defalult:
//			Serializer接口的类的value
			overrideValue(properties, "value.serializer", "cn.tang.tframe.mq.kafka.serial.QtMessageSerializer");//defalult:
			List<ProducerInterceptor> producerInterceptors = SpiUtils.listSpi(ProducerInterceptor.class);
			StringBuffer intecptStr = new StringBuffer();
			if(null != producerInterceptors){
				for(int i = 0;i<producerInterceptors.size();i++){
					if(i != 0){
						intecptStr.append(",");
					}
					intecptStr.append(producerInterceptors.get(i).getClass().getName());
				}
			}
//			用作拦截器的类的列表。
			overrideValue(properties, "interceptor.classes",intecptStr.toString());//defalult:
			return new DelegateKafkaProducer(new KafkaProducer<IQtMessager, IQtMessager>(properties));
		}
		private static void setDefault(Properties properties,String key,Object defalt){
			if(StringUtils.isNotBlank(key) && null != properties && null != defalt){
				Object o = properties.get(key);
				if(null == o){
					o = defalt;
				}
				properties.put(key, o);
			}
		}
		private static void overrideValue(Properties properties,String key,Object newValue){
			if(StringUtils.isNotBlank(key) && null != properties && null != newValue){
				properties.put(key, newValue);
			}
		}
	}

	public static IQtProducer getQtProducer(){
		DelegateKafkaProducer instance = KafkaProducerClientHolder.instance;
		if(null == instance){
			throw new RuntimeException("build kafka producer client exception");
		}
		return instance;
	}


}
